package alilog

import (
	"errors"
	sls "github.com/aliyun/aliyun-log-go-sdk"
	"github.com/astaxie/beego/logs"
	"github.com/astaxie/beego/toolbox"
	"github.com/golang/protobuf/proto"
	"strings"
	"sync"
	"time"
)

var (
	client   sls.ClientInterface
	logQueue map[string][]*sls.Log
	lock     sync.Mutex
)

// Option 阿里云日志配置
type Option struct {
	AccessKeyID     string
	AccessKeySecret string
	Endpoint        string
}

// Init 初始化
func Init(opt Option) {
	logQueue = make(map[string][]*sls.Log)
	client = sls.CreateNormalInterface(opt.Endpoint, opt.AccessKeyID, opt.AccessKeySecret, "")
	lock = sync.Mutex{}
	// 启动定时消费日志服务
	startWrite()
}

func Record(projectName string, logStoreName string, topic string, logData map[string]string) error {
	// 验证
	if projectName == "" {
		logs.Error("projectName is empty")
		return errors.New("projectName is empty")
	}
	if logStoreName == "" {
		logs.Error("logStoreName is empty")
		return errors.New("logStoreName is empty")
	}
	if topic == "" {
		logs.Error("topic is empty")
		return errors.New("topic is empty")
	}
	if logData == nil {
		logs.Error("logData is empty")
		return errors.New("logData is empty")
	}

	content := []*sls.LogContent{}
	for k, v := range logData {
		content = append(content, &sls.LogContent{
			Key:   proto.String(k),
			Value: proto.String(v),
		})
	}
	log := &sls.Log{
		Time:     proto.Uint32(uint32(time.Now().Unix())),
		Contents: content,
	}
	key := getKey(projectName, logStoreName, topic)
	push(key, log)
	return nil
}

func push(key string, log *sls.Log) {
	lock.Lock()
	//_, ok := logQueue[key]
	//if !ok {
	//	logQueue[key] = []*sls.Log{}
	//}
	logQueue[key] = append(logQueue[key], log)
	lock.Unlock()
}

// pop 消费日志
func pop(count int) error {
	sendLogs := make(map[string][]*sls.Log)
	lock.Lock()
	for k, v := range logQueue {
		_count := len(v)
		if _count >= count {
			sendLogs[k] = v[:count]
			logQueue[k] = v[count:]
		} else if _count > 0 {
			sendLogs[k] = v[0:_count]
			delete(logQueue, k)
		}
	}
	//logQueue
	lock.Unlock()
	for k, v := range sendLogs {
		projectName, logStoreName, topic := resolveKey(k)
		loggroup := &sls.LogGroup{
			Topic:  proto.String(topic),
			Source: proto.String(""),
			Logs:   v,
		}
		err := client.PutLogs(projectName, logStoreName, loggroup)
		if err != nil {
			logs.Error("日志消费出错:", err, " topic:", topic)
		}
	}
	return nil
}

// getKey 组装key
func getKey(projectName string, logStoreName string, topic string) string {
	return projectName + "::" + logStoreName + "::" + topic
}

// resolveKey 解析key
func resolveKey(key string) (string, string, string) {
	keys := strings.Split(key, "::")
	if len(keys) == 3 {
		return keys[0], keys[1], keys[2]
	}
	return "", "", ""
}

// startWrite 启动定时消费日志服务
func startWrite() {
	n := toolbox.NewTask("liupei-push-log", "*/5 * * * * *", func() error {
		// 每次处理10条
		go pop(10)
		//logs.Info(len(logQueue))
		return nil
	})
	toolbox.AddTask("liupei-push-log", n)
	toolbox.StartTask()
}
